
#undef LOGGER
#define NDEBUG

#include <boost/scoped_ptr.hpp>
#include <xsocket/active_reactor.hpp>

// -*-mode:c++; coding:utf-8-*-

#ifndef _HANDLER_STATUS_
#define _HANDLER_STATUS_

struct handler_status
{
	unsigned int m_left_in_message;
	unsigned int m_left_out_message;

	handler_status()
		: m_left_in_message(0)
		, m_left_out_message(0) {}
};

#endif	// _HANDLER_STATUS_

// -*-mode:c++; coding:utf-8-*-


#ifndef _CLIENT_MESSAGE_HANDLER_HPP_
#define _CLIENT_MESSAGE_HANDLER_HPP_

#include <utility>
#include <sstream>
#include <string>
#include <boost/shared_ptr.hpp>
#include <boost/thread/condition.hpp>
#include <boost/thread/thread.hpp>

#include <xthread/thread.hpp>
#include <appframe/circular_array_queue.hpp>
#include <xsocket/bytebuffer.hpp>

#include <stringable.hpp>

// struct client_message 
// {
// 	uint32_t m_connection_id;
// 	Header *m_packet;

// 	client_message(uint32_t p_id, Header *p_packet)
// 		: m_connection_id(p_id)
// 		, m_packet(p_packet) {		
// 	}
		
// };

class ClientMessageHandler : public Thread,
			     public stringable
{
public:
	typedef std::pair<uint32_t, ByteBuffer*> MessageType;
	typedef CircularArrayQueue<MessageType> MessageQueue;

public:

	ClientMessageHandler (unsigned int id,
			      handler_status &status);

	virtual ~ClientMessageHandler () {}

        virtual void run ();

	// 供外部对输入和输出队列进行操作的接口
	void sendMessage (const MessageType &msg);
	MessageType recvMessage ();

	// for debug
	std::string to_string () const {
		std::stringstream out;
		out << "{input_queue_size:" << _input_queue->size ()
		    << "; output_queue_size:" << _output_queue->size ()
		    << "}";
		return out.str();
	}

protected:
	// 一直等待、重试，直到成功
	MessageType popFromInputUntilSuccess ();
	void pushToOutputUntilSuccess (const MessageType &msg);

	void do_read(const MessageType &p_msg);
	void do_write(const MessageType &p_msg);
	void do_scan_open(const MessageType &p_msg);
	void do_scan_next(const MessageType &p_msg);
	void do_scan_close(const MessageType &p_msg);
	void do_unexpected(const MessageType &p_msg);
	void do_list_tablet(const MessageType &p_msg);

private:
	boost::shared_ptr<MessageQueue> _input_queue;
	boost::mutex m_input_queue_monitor;
	boost::condition m_input_queue_not_empty;
	
	boost::shared_ptr<MessageQueue> _output_queue;
	boost::mutex m_output_queue_monitor;
	boost::condition m_output_queue_not_full;
	
	handler_status &m_status;

	// for debug & profile
	unsigned int _id;
	unsigned int _input_count;
	unsigned int _output_count;
};

#endif	// _CLIENT_MESSAGE_HANDLER_HPP_
// -*-mode:c++; coding:utf-8-*-

#include <iostream>
#include <utility>
#include <memory>
using namespace std;

#include <xbase_exception.hpp>
#include <xbase_datagram_factory.hpp>

ClientMessageHandler::ClientMessageHandler (unsigned int id,
					    handler_status &status)
	: Thread()
	, _input_queue()
	, _output_queue()
	, m_status(status)
	, _id(id)
	, _input_count(0)
	, _output_count(0)
{
	_input_queue = boost::shared_ptr<MessageQueue>(new MessageQueue(102400));
	_output_queue = boost::shared_ptr<MessageQueue>(new MessageQueue(102400));
}

void ClientMessageHandler::sendMessage (const ClientMessageHandler::MessageType &msg) {
	// 这里有重试及超时机制

	int ret = 0;

	int retry = 3;
	do {
		m_status.m_left_in_message = _input_queue->size();
		if(_input_queue->full()) {
			// 超时
			if(retry <= 0) {
				LOG(WARN, "input message queue in client message handler "
				    << _id << " " << *this << " is always full, break." );
				throw QueueFull ("input queue is full");
			}

			// 未超时
			LOG(WARN, "input message queue in client message handler "
			    << _id << " " << *this << " is full,"
			    << " wait for a while("
			    << 3
			    << "ms) and retry("
			    << retry << ")");
			
			--retry;
			this->sleep(300); // 毫秒ms
		} else {
			ret = this->_input_queue->push (msg);
			m_input_queue_not_empty.notify_one();
			m_status.m_left_in_message = _input_queue->size();
			
			// 应该会成功的
			LOG_IF((ret != 0),
				WARN,
			       "ClientMessageHandler::sendMessage insert message error,"
			       " but should not occur error here.");
			
			// success
			break;
		}
	} while (true);
}

ClientMessageHandler::MessageType ClientMessageHandler::recvMessage () {
	if (this->_output_queue->empty ())
	{
		m_status.m_left_out_message = 0;
		return ClientMessageHandler::MessageType(0,0);
	}
	else
	{
		MessageType msg = _output_queue->front ();
		_output_queue->pop ();
		m_output_queue_not_full.notify_one();
		m_status.m_left_out_message = _output_queue->size();
		return msg;
	}
}

void ClientMessageHandler::run () {
	LOG(TRACE, "client message handler " << _id << " started.");

	while (true) {
		MessageType msg;
		// get a message
		msg = popFromInputUntilSuccess ();
		++ _input_count;
			
		LOG(TRACE, "in hanlder " << *this
		    << " got " << msg.second->remaining() << "bytes message");
		
		// decode it
		int type = XBaseDatagramFactory::getType (*(msg.second));
		switch (type) {
		case READ_REQ:
			this->do_read(msg);
			break;
		case WRITE_REQ:
			this->do_write(msg);
			break;
		case SCAN_OPEN_REQ:
			this->do_scan_open(msg);
			break;
		case SCAN_NEXT_REQ:
			this->do_scan_next(msg);
			break;
		case SCAN_CLOSE_REQ:
			this->do_scan_close(msg);
			break;
		case LIST_TABLET_REQ:
			this->do_list_tablet(msg);
			break;
		default:
			this->do_unexpected(msg);
			break;
		}
	}

	LOG(INFO, "client message handler " << _id << " quit.");
}

ClientMessageHandler::MessageType
ClientMessageHandler::popFromInputUntilSuccess () 
{
	//TIME_EVALUATOR("ClientMessageHandler::popFromInputUntilSuccess ", "used ");
	
	m_status.m_left_in_message = _input_queue->size();
	if(_input_queue->empty())
	{
		boost::mutex::scoped_lock lock(m_input_queue_monitor);
		while (_input_queue->empty ())
		{
			LOG(TRACE, "ClientMessageHandler::popFromInputUntilSuccess input queue is empty in handler #"
			    << _id << ", waiting for message ... ");
			m_input_queue_not_empty.wait(lock);
			LOG(TRACE, "ClientMessageHandler::popFromInputUntilSuccess got notified");
		}
	}

	MessageType msg = _input_queue->front ();
	_input_queue->pop ();
	m_status.m_left_in_message = _input_queue->size();

	return msg;
}

void ClientMessageHandler::pushToOutputUntilSuccess (const MessageType &msg) {
	//TIME_EVALUATOR("ClientMessageHandler::pushToOutputUntilSuccess ", "used ");
	
	m_status.m_left_out_message = _output_queue->size();
	if(_output_queue->full())
	{
		boost::mutex::scoped_lock lock(m_output_queue_monitor);
		while(_output_queue->full())
		{
			LOG(WARN, "ClientMessageHandler::pushToOutputUntilSuccess output queue is full in handler #"
			    << _id << ", waiting for message ... ");
			m_output_queue_not_full.wait(lock);
			LOG(TRACE, "ClientMessageHandler::pushToOutputUntilSuccess got notified");
		}
	}
	const int ret = _output_queue->push (msg);
	m_status.m_left_out_message = _output_queue->size();
	LOG_IF((ret != 0),
	       WARN,
	       "ClientMessageHandler::pushToOutputUntilSuccess push failed, ret=" << ret);
	(void)ret;
}

// 
// the following function should never throws
//
// TODO: 现在出错后只简单的将连接关闭，以后可能做的完善一些：解码失败的
// 将连接关闭；解码成功的返回失败
// 

// 调试模式下，不捕获异常，以便暴露问题。
#ifndef NDEBUG
#	define try if(true)
#	define catch(x) else
#endif	// NDEBUG

void ClientMessageHandler::do_read(const MessageType &p_msg)
{
	LOG(TRACE, "ClientMessageHandler::do_read");

	std::auto_ptr<ByteBuffer> _holder(p_msg.second);
	MessageType _resp_msg = p_msg;

	try {
		ReadReq _req;
		XBaseDatagramFactory::decode(*(p_msg.second), _req);

		ReadResp _resp;
	
		//boost::shared_ptr<ByteBuffer> _buffer(new ByteBuffer(MAX_PACKET_SIZE));
		//_resp_msg = p_msg;//(p_msg.first, _buffer);
		_resp_msg.second->clear();
		XBaseDatagramFactory::encode(*_resp_msg.second, _resp);

		pushToOutputUntilSuccess(_resp_msg);
		_holder.release();
	} catch(.../*const std::exception &e*/) {
		LOG(ERROR, "ClientMessageHandler::do_read throw exception of "
		    // << typeid(e).name()
		    << "..., close this socket");
		// 通过发送NULL包，关闭这个连接
		MessageType close_resp(p_msg.first, 0);
		pushToOutputUntilSuccess(close_resp);
	}
}

void ClientMessageHandler::do_write(const MessageType &p_msg)
{
	LOG(TRACE, "ClientMessageHandler::do_write");

	std::auto_ptr<ByteBuffer> _holder(p_msg.second);
	MessageType _resp_msg = p_msg;//(p_msg.first, _buffer);

	try {
		// WriteReq _req;
		// XBaseDatagramFactory::decode(*(p_msg.second), _req);

		//const write_request_wrapper *req = reinterpret_cast<const write_request_wrapper*>(p_msg.second->array());
		
		WriteResp _resp;
		
		//boost::shared_ptr<ByteBuffer> _buffer(new ByteBuffer(MAX_PACKET_SIZE));
		//MessageType _resp_msg = p_msg;//(p_msg.first, _buffer);
		_resp_msg.second->clear();
		XBaseDatagramFactory::encode(*_resp_msg.second, _resp);
	
		pushToOutputUntilSuccess(_resp_msg);
		_holder.release();
	} catch(.../*const std::exception &e*/) {
		LOG(ERROR, "ClientMessageHandler::do_write throw exception of "
		    // << typeid(e).name()
		    << "..., close this socket");
		// 通过发送NULL包，关闭这个连接
		MessageType close_resp(p_msg.first, 0);
		pushToOutputUntilSuccess(close_resp);
	}
}

void ClientMessageHandler::do_scan_open(const MessageType &p_msg)
{
	LOG(TRACE, "ClientMessageHandler::do_scan_open");

	std::auto_ptr<ByteBuffer> _holder(p_msg.second);
	MessageType _resp_msg = p_msg;//(p_msg.first, _buffer);

	try {
		ScanOpenReq _req;
		XBaseDatagramFactory::decode(*(p_msg.second), _req);

		LOG(TRACE, "ClientMessageHandler::do_scan_open req: " << _req.to_string());
		
		ScanOpenResp _resp;
		
		//boost::shared_ptr<ByteBuffer> _buffer(new ByteBuffer(MAX_PACKET_SIZE));
		//MessageType _resp_msg = p_msg;//(p_msg.first, _buffer);
		_resp_msg.second->clear();
		XBaseDatagramFactory::encode(*_resp_msg.second, _resp);
	
		pushToOutputUntilSuccess(_resp_msg);
		_holder.release();
	} catch(.../*const std::exception &e*/) {
		LOG(ERROR, "ClientMessageHandler::do_scan_open throw exception of "
		    // << typeid(e).name()
		    << "..., close this socket");
		// 通过发送NULL包，关闭这个连接
		MessageType close_resp(p_msg.first, 0);
		pushToOutputUntilSuccess(close_resp);
	}
}

void ClientMessageHandler::do_scan_next(const MessageType &p_msg)
{
	LOG(TRACE, "ClientMessageHandler::do_scan_next");

	std::auto_ptr<ByteBuffer> _holder(p_msg.second);
	MessageType _resp_msg = p_msg;//(p_msg.first, _buffer);
	
	try {
		ScanNextReq _req;
		XBaseDatagramFactory::decode(*(p_msg.second), _req);

		ScanNextResp _resp;

		//boost::shared_ptr<ByteBuffer> _buffer(new ByteBuffer(MAX_PACKET_SIZE));
		//MessageType _resp_msg = p_msg;//(p_msg.first, _buffer);
		_resp_msg.second->clear();
		XBaseDatagramFactory::encode(*_resp_msg.second, _resp);

		pushToOutputUntilSuccess(_resp_msg);
		_holder.release();
	} catch(.../*const std::exception &e*/) {
		LOG(ERROR, "ClientMessageHandler::do_scan_next throw exception of "
		    //<< typeid(e).name()
		    << "..., close this socket");
		// 通过发送NULL包，关闭这个连接
		MessageType close_resp(p_msg.first, 0);
		pushToOutputUntilSuccess(close_resp);
	}
}

void ClientMessageHandler::do_scan_close(const MessageType &p_msg)
{
	TRACE("ClientMessageHandler::do_scan_close");
	TIME_EVALUATOR("ClientMessageHandler::do_scan_close ", "used ");

	std::auto_ptr<ByteBuffer> _holder(p_msg.second);
	MessageType _resp_msg = p_msg;//(p_msg.first, _buffer);

	try {
		////ScanCloseReq _req;
		////XBaseDatagramFactory::decode(*(p_msg.second), _req);
		
		ScanCloseResp _resp;
		
		//boost::shared_ptr<ByteBuffer> _buffer(new ByteBuffer(MAX_PACKET_SIZE));
		//MessageType _resp_msg = p_msg;//(p_msg.first, _buffer);
		_resp_msg.second->clear();
		XBaseDatagramFactory::encode(*_resp_msg.second, _resp);

		pushToOutputUntilSuccess(_resp_msg);
		_holder.release();
	} catch(.../*const std::exception &e*/) {
		LOG(ERROR, "ClientMessageHandler::do_scan_close throw exception of "
		    // << typeid(e).name()
		    << "..., close this socket");
		// 通过发送NULL包，关闭这个连接
		MessageType close_resp(p_msg.first, 0);
		pushToOutputUntilSuccess(close_resp);
	}
}

void ClientMessageHandler::do_unexpected(const MessageType &p_msg)
{
	LOG(WARN, "ClientMessageHandler::do_unexpected unexpected message type 0x"
	    << std::hex << std::uppercase
	    << XBaseDatagramFactory::getType(*(p_msg.second))
	    << ".");

	delete p_msg.second;

	// 通过发送NULL包，关闭这个连接
	MessageType close_resp(p_msg.first, 0);
	pushToOutputUntilSuccess(close_resp);
}

void ClientMessageHandler::do_list_tablet(const MessageType &p_msg)
{
	LOG(TRACE, "ClientMessageHandler::do_list_tablet");

	std::auto_ptr<ByteBuffer> _holder(p_msg.second);
	MessageType _resp_msg = p_msg;//(p_msg.first, _buffer);

	try {
		list_tablet_resp _resp;
		
		//boost::shared_ptr<ByteBuffer> _buffer(new ByteBuffer(MAX_PACKET_SIZE));
		//MessageType _resp_msg = p_msg;//(p_msg.first, _buffer);
		_resp_msg.second->clear();
		XBaseDatagramFactory::encode(*_resp_msg.second, _resp);

		pushToOutputUntilSuccess(_resp_msg);
		_holder.release();
	} catch(.../*const std::exception &e*/) {
		LOG(ERROR, "ClientMessageHandler::do_list_tablet throw exception of "
		    // << typeid(e).name()
		    << "..., close this socket");
		// 通过发送NULL包，关闭这个连接
		MessageType close_resp(p_msg.first, 0);
		pushToOutputUntilSuccess(close_resp);
	}
}
// -*-c++-*-


#ifndef _CLIENT_MESSAGE_PROCESSOR_HPP_
#define _CLIENT_MESSAGE_PROCESSOR_HPP_

#include <stdint.h>
#include <ext/hash_map>
#include <ext/hash_set>
#include <utility>

#include <xsocket/imessageprocessor.hpp>
#include <xsocket/connectioninfo.hpp>

#include <xbase_datagram_factory.hpp>
#include <xbase_utility.hpp>
#include <stringable.hpp>

class ClientMessageProcessor : public IMessageProcessor
			     , public stringable
{

public:
	// typedefs
	typedef __gnu_cxx::hash_map<uint32_t, ConnectionInfo* > ConnectionMap;
	typedef __gnu_cxx::hash_set<ConnectionInfo* > ConnectionSet;

	typedef ClientMessageHandler::MessageType MessageType;

public:

	ClientMessageProcessor ();

	virtual ~ClientMessageProcessor ();

	virtual void accept (ConnectionInfo &conn);
	virtual void remove (ConnectionInfo &conn);

	virtual void process_input (ConnectionInfo &conn, ByteBuffer &bb);
	virtual void process_output ();

	virtual int frame (ByteBuffer &bb);

	// for debug
	std::string to_string () const {
		std::stringstream out;
		out << "{handlerNumber:" << _handlerNumber
		    << "; connections_size:" << _connections.size ()
		    << "; connections_by_id_size:" << _connections_by_id.size ()
		    << "}";
		
		return out.str();
	}

private:
	// TODO: 用线程池来实现多个Handler
	
	// 客户端请求的处理线程组
	unsigned int _handlerNumber;
	ClientMessageHandler **_handlers;
	handler_status *m_handler_status;
	unsigned int m_cur_handler;

	ConnectionSet _connections;
	ConnectionMap _connections_by_id;
	// unsigned int _max_connetions;

	// for debug & profile
	unsigned int _input_count;
	unsigned int _sent_input_count;
	unsigned int _output_count;
	unsigned int _sent_output_count;
};

#endif	// _CLIENT_MESSAGE_PROCESSOR_HPP_
// -*-mode:c++; coding:utf-8-*-

#include <exception>
#include <utility>
#include <algorithm>
using namespace std;

#include <boost/foreach.hpp>

#include <xsocket/bytebuffer.hpp>

#include <xbase_exception.hpp>
#include <xbase_utility.hpp>
#include <logger.hpp>

ClientMessageProcessor::ClientMessageProcessor ()
	: _handlerNumber(10)
	, _handlers(NULL)
	, m_handler_status(0)
	, m_cur_handler(0)
	, _input_count(0)
	, _output_count(0)
{
	_handlers = new ClientMessageHandler* [_handlerNumber];
	m_handler_status = new handler_status[_handlerNumber];

	// 初始化
	for (unsigned int i=0; i<_handlerNumber; ++i)
	{
		_handlers [i] = NULL;
	}
	for (unsigned int i=0; i<_handlerNumber; ++i)
	{
		_handlers [i] = new ClientMessageHandler(i, m_handler_status[i]);
		_handlers [i]->start();
	}
	LOG(TRACE, _handlerNumber << " client message handlers started.");
}


ClientMessageProcessor::~ClientMessageProcessor () {
	if (_handlers != NULL)
	{
		for (unsigned int i=0; i<_handlerNumber; ++i)
		{
			if (_handlers [i] != NULL)
			{
				delete _handlers [i];
				_handlers [i] = NULL;
			}
		}
		
		delete [] _handlers;
		_handlers = NULL;
	}

	// first save connection in list
	std::vector<ConnectionInfo*> _list;
	BOOST_FOREACH(ConnectionInfo *_p, _connections)
	{
		_list.push_back(_p);
	}
	// then close all alive connection
	BOOST_FOREACH(ConnectionInfo *_p, _list)
	{
		_p->close();
	}
}

void ClientMessageProcessor::process_input (ConnectionInfo &conn, ByteBuffer &bb) {
	TRACE("ClientMessageProcessor::process_input");
	TIME_EVALUATOR("ClientMessageProcessor::process_input ", " used ");

	// update load info

	// for debug
	++ _input_count;

	int conn_id = conn.get_id ();

	// 拷贝到新的bytebuffer
	//boost::shared_ptr<ByteBuffer> buffer(new ByteBuffer(bb.remaining()));
	std::auto_ptr<ByteBuffer> buffer(new ByteBuffer(std::max((int)MAX_PACKET_SIZE, bb.remaining())));
	buffer->put(bb);
	buffer->flip();
	// ByteBuffer::arraycopy (bb.array (), bb.position (),
	// 		       buffer->array (), buffer->position (),
	// 		       bb.remaining ());
	// buffer->limit(bb.remaining ());
	
	MessageType msg(conn_id, buffer.get());

	// 选择一个处理线程进行处理
	// TODO: 用更好的选择策略，当前的策略是同一个Socket的数据被放到相同的处理线程
	//_handlers [conn_id % _handlerNumber]->sendMessage (msg);
	_handlers[m_cur_handler]->sendMessage (msg);
	buffer.release();
	
	++m_cur_handler;
	if(m_cur_handler >= _handlerNumber) {
		m_cur_handler = 0;
	}

	// for debug
	++ _sent_input_count;
	
}

void ClientMessageProcessor::process_output () {
	// for load info
	unsigned int left_in_message = 0;
	unsigned int left_out_message = 0;

	//MessageType msg;
	unsigned int count_per_handler = 0;

	for (unsigned int i=0; i<_handlerNumber; ++i) {
		count_per_handler = 0;
		for (int j=0; j<128; ++j) {
			MessageType msg = _handlers [i]->recvMessage ();

			if ((msg.first == 0) && (msg.second == 0)) // no message
				break;

			TRACE("ClientMessageProcessor::process_output");
			TIME_EVALUATOR("ClientMessageProcessor::process_output ", " used ");
	
			// for debug
			// 记录成功取出及总的的响应包数，用于统计。
			++ count_per_handler;
			++ _output_count;
			
			boost::scoped_ptr<ByteBuffer> buffer(msg.second);
			// 准备发送数据包，打印日志也需要先 flip
			if (buffer)
				buffer->flip ();
			
			ConnectionMap::iterator iter = _connections_by_id.find (msg.first);
			
			// 没有该连接，可能是处理期间已经关闭或
			// 未知的连接，直接将其丢弃。
			if (iter == _connections_by_id.end()) {
				ostringstream stream;
				if (buffer)
					stream << CommandName(XBaseDatagramFactory::getType(*buffer));
				else
					stream << "NULL";
				
				LOG(ERROR, "response message("
				    << stream.str()
				    << ") to client on a"
				    << " CLOSED or UNKNOWN connection, drop it.");
				
				continue;
			}
			
			// 通常情况下，连接断开就会从列表中删除
			// 的，所以这里取到的一定是存活的连接，
			// 但以防未知的错误发生，这里进行了一次
			// 检查
			
			// 若是已关闭连接，则不应该出现在该列表
			// 中，检测出来后将其删除，并将相关的响
			// 应包丢弃
			if(! iter->second->is_open()) {
				ostringstream stream;
				if (buffer)
					stream << CommandName(XBaseDatagramFactory::getType(*buffer));
				else
						stream << "NULL";
				
				LOG(ERROR, "response message("
				    << stream.str()
					    <<") to client on a"
				    << " CLOSED connection, drop it.");
				
				this->remove(*iter->second); // must first
				iter->second->close(); // must second
				continue;
			}
			
			// 空的数据指针，则将连接关闭
			if (! buffer) {
				LOG(INFO, "got a NULL response buffer on connection "
				    << *iter->second << ", so close this connection.");

				// 前面已经保证该连接存在并没有
				// 关闭，因此只调用close即可
				iter->second->close();
				continue;
			}
			
			// 若是一个空的数据包，也将连接关闭
			if (buffer->remaining () == 0) {
				LOG(INFO, "got a EMPTY response buffer, so close the connection.");
				
				iter->second->close();
				continue;
			}
			
			// 若没准备好写，则可能是上次的没有发送
			// 完，或连接出问题了，尝试将其flush一
			// 下再作处理
			if(! iter->second->ready_to_write()) {
				LOG(TRACE, "flush on connection " << *iter->second << ".");
				iter->second->flush ();
			}
			
			// 若仍然不能写，很可能是出问题了，也将
			// 其关闭。
			if(! iter->second->ready_to_write()) {
				LOG(ERROR, "connection " << *iter->second
				    << " not ready to write, so close it."
				    << "(should me?)");
				
				iter->second->close ();
				continue;
			}

			// 到现在才排除了各种可能的问题和错误，开始发送包
			int ret = iter->second->write (*buffer);
			
			// 若没发送成功，则将连接关闭
			if(ret < 0) {
				LOG(ERROR, "write to connection "
				    << *iter->second
				    << " failed, so close it.");
				
				// 关闭操作会调用自身的remove函
				// 数将其删除
				iter->second->close ();
				continue;
			}
			
			// 到此才发送成功，设置统计用的计数器
			++ _sent_output_count;
		}
		
		// get load info
		left_in_message += m_handler_status[i].m_left_in_message;
		left_out_message += m_handler_status[i].m_left_out_message;
	}
}

void ClientMessageProcessor::accept (ConnectionInfo &conn) {
	// DONE: 这里不需要考虑插入失败吗？虽然基本上不会有两个相同的
	// Key，除非是同一个对象。而这时候它已经在里面了，就不用再插入
	// 了吧。
	_connections.insert (&conn);
	_connections_by_id.insert (ConnectionMap::value_type(conn.get_id(), &conn));

	LOG(INFO, "a connection from " << conn << " accepted.");
}

void ClientMessageProcessor::remove (ConnectionInfo &conn) {
	_connections.erase (&conn);
	_connections_by_id.erase (conn.get_id());

	LOG(INFO, "a connection from " << conn << " closed.");
}

int ClientMessageProcessor::frame (ByteBuffer &bb) {
	return XBaseDatagramFactory::frame (bb);
}

int main(int argc, char **argv)
{
	ClientMessageProcessor processor;
	ActiveReactor reactor;
	reactor.init();
	const int ret = reactor.add_acceptor("0.0.0.0", 20083, processor);
	if(ret < 0)
	{
		return -1;
	}

	while(true)
	{
		reactor.poll();

		// wasting time
		int n=9;
		time_t time = cur_micro_seconds();
		while(n--)
		{
			time += cur_micro_seconds();
		}
	}

	return 0;
}
